/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.dpl.data.mmloader.globalParas;

import cn.ac.iie.dpl.data.mmloader.util.ConfigureUtil;
import iie.metastore.MetaStoreClient;
import iie.mm.client.ClientAPI;
import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.log4j.Logger;

/**
 *
 * @author iie
 */
public class GlobalParas {

    public static Logger log = Logger.getLogger(GlobalParas.class.getName());
    public static String region = null;
    public static AtomicBoolean isShouldExit = new AtomicBoolean(false);
    public static MetaStoreClient metastoreCli = null;
    public static int maxMessageSize = 4096;
    public static String machineIP = null;
    public static String nameServer = null;
    public static int batchSize = 3000;
    public static ConcurrentHashMap<String, Map<String, ArrayList<ClientAPI>>> ClientAPIMap = new ConcurrentHashMap(10, 0.8F);
    public static List<String> srcSchemaName = new ArrayList<String>();
    public static ConcurrentHashMap<String, String> srcDesScheMap = new ConcurrentHashMap<String, String>();
    public static ConcurrentHashMap<String, String> srcDesTopicMap = new ConcurrentHashMap<String, String>();
    public static ConcurrentHashMap<String, String> srcTopicMap = new ConcurrentHashMap<String, String>();
    public static ConcurrentHashMap<String, String> srcTimeColumnNameMap = new ConcurrentHashMap<String, String>();
    public static ConcurrentHashMap<String, Integer> srcSchemaThreadSizeMap = new ConcurrentHashMap<String, Integer>();
    public static ConcurrentHashMap<String, List<String>> srcSchemaColumnNameMap = new ConcurrentHashMap<String, List<String>>();
    public static ConcurrentHashMap<String, LinkedBlockingQueue<GenericRecord>> srcSchemaQueneMap = new ConcurrentHashMap<String, LinkedBlockingQueue<GenericRecord>>();
    public static ConcurrentHashMap<String, Map<Utf8, Utf8>> srcToDocDesc = new ConcurrentHashMap<String, Map<Utf8, Utf8>>();
    public static ConcurrentHashMap<String, Map<String, String>> srcToDesDocDesc = new ConcurrentHashMap<String, Map<String, String>>();
    /* */
    public static Schema dpldocsSchema = null;
    public static Schema dpldocSchema = null;

    public static void init() throws Exception {
        ConfigureUtil cfg = new ConfigureUtil();
        nameServer = cfg.getProperty("nameServer");
        String[] schemaThreads = cfg.getProperty("schemaThread").split("[;]");
        String[] srcSchemaNames = cfg.getProperty("srcSchemaName").split("[;]");
        String[] desSchemaName = cfg.getProperty("desSchemaName").split("[;]");
        String[] schemaQueueSizes = cfg.getProperty("schemaQueueSize").split("[;]");
        String[] columns = cfg.getProperty("columnName").split("[;]");
        String[] topics = cfg.getProperty("srcTopic").split("[;]");
        String[] srctimeColumnNames = cfg.getProperty("timeColumnName").split("[;]");
        String[] desTopics = cfg.getProperty("desTopic").split("[;]");
        maxMessageSize = cfg.getIntProperty("maxMessageSize");
        batchSize = cfg.getIntProperty("batchSize");
        for (int i = 0; i < srcSchemaNames.length; i++) {
            srcSchemaName.add(srcSchemaNames[i]);
            srcTopicMap.put(srcSchemaNames[i], topics[i]);
            srcDesScheMap.put(srcSchemaNames[i], desSchemaName[i]);
            srcSchemaThreadSizeMap.put(srcSchemaNames[i], Integer.parseInt(schemaThreads[i]));
            srcSchemaColumnNameMap.put(srcSchemaNames[i], Arrays.asList(columns[i].split("[,]")));
            srcSchemaQueneMap.put(srcSchemaNames[i], new LinkedBlockingQueue<GenericRecord>(Integer.parseInt(schemaQueueSizes[i])));
            srcTimeColumnNameMap.put(srcSchemaNames[i], srctimeColumnNames[i]);
            srcDesTopicMap.put(srcSchemaNames[i], desTopics[i]);
        }
        try {
            GlobalParas.metastoreCli = new MetaStoreClient(cfg.getProperty("metaStoreCient").split("[:]")[0], Integer.parseInt(cfg.getProperty("metaStoreCient").split("[:]")[1]));
        } catch (MetaException ex) {
            log.error(ex, ex);
            return;
        }
        machineIP = cfg.getProperty("machineIP");

        GlobalParas.region = cfg.getProperty("region");

        Protocol protocold = Protocol.parse(new File("docs.json"));
        dpldocsSchema = protocold.getType("docs");
        Protocol protocoldoc = Protocol.parse(new File("doc.json"));
        dpldocSchema = protocoldoc.getType("doc");
    }
}
